﻿// aac_server.cpp : 此文件包含 "main" 函数。程序执行将在此处开始并结束。
//

#include "pch.h"
#include <iostream>
#include <set>
#include <list>
#include "websocketpp/config/asio_no_tls.hpp"
#include "websocketpp/server.hpp"
#include "websocketpp/common/thread.hpp"

#include "nlohmann/json.hpp"


typedef websocketpp::server<websocketpp::config::asio> server;
using websocketpp::connection_hdl;
using websocketpp::lib::placeholders::_1;
using websocketpp::lib::placeholders::_2;
using websocketpp::lib::bind;

using websocketpp::lib::thread;
using websocketpp::lib::mutex;
using websocketpp::lib::lock_guard;
using websocketpp::lib::unique_lock;
using websocketpp::lib::condition_variable;


class BytesTools {
public:
	static int byteArrayToInt(const unsigned char* byteArray) {
		return (byteArray[0] << 24) | (byteArray[1] << 16) | (byteArray[2] << 8) | byteArray[3];
	}

	//int main() {
	//	unsigned char byteArray[] = { 0x00, 0x00, 0x00, 0x01 };
	//	int result = byteArrayToInt(byteArray);

	//	std::cout << "Converted Integer: " << result << std::endl;

	//	return 0;
	//}

};


enum action_type {
	SUBSCRIBE,
	UNSUBSCRIBE,
	MESSAGE
};

struct action {
	action(action_type t, connection_hdl h) : type(t), hdl(h) {}
	action(action_type t, connection_hdl h, server::message_ptr m)
		: type(t), hdl(h), msg(m) {}

	action_type type;
	websocketpp::connection_hdl hdl;
	server::message_ptr msg;
};


class ACCServer {

public:

	ACCServer() {

		m_server.set_reuse_addr(true);
		m_server.init_asio();
		m_server.clear_access_channels(websocketpp::log::alevel::frame_header);
		m_server.clear_access_channels(websocketpp::log::alevel::control);


		m_server.set_open_handler(bind(&ACCServer::on_open, this, ::_1));
		m_server.set_close_handler(bind(&ACCServer::on_close, this, ::_1));
		m_server.set_message_handler(bind(&ACCServer::on_message, this, ::_1, ::_2));
		m_server.set_ping_handler(bind(&ACCServer::on_ping, this, ::_1, ::_2));
		m_server.set_pong_handler(bind(&ACCServer::on_pong, this, ::_1, ::_2));
	
	}


	void run(std::uint16_t port)
	{
		m_server.listen(port);
		m_server.start_accept();

		try {
			m_server.run();
		}
		catch (const std::exception& e)
		{
			std::cout << e.what() << std::endl;
		}
	}

	bool on_ping(connection_hdl hdl, std::string s1)
	{
		//std::cout << "ping " << std::endl;
		this->m_server.pong(hdl, s1);
		return true;
	}

	void on_pong(connection_hdl, std::string s1)
	{
		std::cout << "pong" << std::endl;
	}

	void on_open(connection_hdl hdl) {
	/*	{
			std::lock_guard<std::mutex>	lock(mListTimeoutLock);
			mListTimeout[hdl] = time(nullptr);
		}*/
		{
			lock_guard<mutex> guard(m_action_lock);
			std::cout << "on_open" << std::endl;
			m_actions.push(action(SUBSCRIBE, hdl));
		}
		m_action_cond.notify_one();
	}

	void on_close(connection_hdl hdl) {
		{
	/*		{std::lock_guard<std::mutex>	lock(mListTimeoutLock);
			if (mListTimeout.count(hdl) > 0)
			{
				mListTimeout.erase(hdl);
			}
			}*/
			lock_guard<mutex> guard(m_action_lock);
			std::cout << "on_close" << std::endl;
			m_actions.push(action(UNSUBSCRIBE, hdl));
		}
		m_action_cond.notify_one();
	}

	void on_message(connection_hdl hdl, server::message_ptr msg) {
		// queue message up for sending by processing thread
		/*{
			std::lock_guard<std::mutex>	lock(mListTimeoutLock);
			mListTimeout[hdl] = time(0);
		}*/
		{
			lock_guard<mutex> guard(m_action_lock);
			//std::cout << "on_message" << std::endl;
			m_actions.push(action(MESSAGE, hdl, msg));
		}
		m_action_cond.notify_one();
	}
	void process_messages() {
		while (1) {
			unique_lock<mutex> lock(m_action_lock);

			while (m_actions.empty()) {
				m_action_cond.wait(lock);
			}

			if (m_actions.size() == 0)
				continue;

			action a = m_actions.front();
			m_actions.pop();

			lock.unlock();

			if (a.type == SUBSCRIBE) {
				//lock_guard<mutex> guard(m_connection_lock);
				m_connections.insert(a.hdl);
				{
					lock_guard<mutex> guard(m_connection_lock);
				}
			}
			else if (a.type == UNSUBSCRIBE) {
				//lock_guard<mutex> guard(m_connection_lock);
				try {
					m_server.close(a.hdl, websocketpp::close::status::normal, "");
				}
				catch (const std::exception& e)
				{
					std::cout << e.what() << std::endl;
				}
				m_connections.erase(a.hdl);
			}
			else if (a.type == MESSAGE) {
				//lock_guard<mutex> guard(m_connection_lock);

				websocketpp::frame::opcode::value opt = a.msg.get()->get_opcode();
				if (opt == websocketpp::frame::opcode::binary)
				{
					parse_data(a);
				}
				else if (opt == websocketpp::frame::opcode::text)
				{
					std::cout << "text:" << a.msg.get()->get_payload() << std::endl;
				}
				else {
					std::cout << a.msg << "  size:" << a.msg.get()->get_payload() << std::endl;
				}

				/*con_list::iterator it;
				for (it = m_connections.begin(); it != m_connections.end(); ++it) {
					m_server.send(*it, a.msg);
				}*/
			}
			else {
				// undefined.
			}
		}
	}

	/**
	 * 1,注册 ，2 开始推流 data是ip列表，并且创建一个流id，3，音频数据 data是aac数据，4结束推流
	 */
	bool parse_data(action& act) {
		const std::string& payload = act.msg.get()->get_payload();
		int size = payload.size();
		if (size < 40)
		{
			std::cout << "valid size" << std::endl;
			on_close(act.hdl);
			return false;
		}

		std::uint8_t* pData = (std::uint8_t*)payload.c_str();

		int cmd;
		int index = 0;
		cmd = BytesTools::byteArrayToInt(pData + index);
		index += 4;
		char stream_id[33] = { 0 };
		memcpy(stream_id, pData + index, 32);
		index += 32;
		int length;
		length = BytesTools::byteArrayToInt(pData + index);
		index += 4;

		std::unique_ptr<std::uint8_t[]> ptrData;
		if (length > 0)
		{
			ptrData.reset(new std::uint8_t[length]);
			memcpy(ptrData.get(), pData + index, length);
		}

		if (cmd == 3)
		{
			//音频数据
			send_boardcast(act, stream_id);
			//	std::cout << "audio received " << std::endl;
		}
		else if (cmd == 1)
		{
			//注册 {"ip":"10.0.3.173","nailId":"0001"}
			std::string register_msg;
			register_msg.append((char*)ptrData.get(), length);
			auto r1 = nlohmann::json::parse(register_msg);
			auto ip = r1["ip"];
			auto nailId = r1["nailId"];
			mRegisterList[ip] = act.hdl;
			std::cout << "register:" << register_msg << "ip:" << ip << " nailId:" << nailId << std::endl;
		}
		else if (cmd == 2)
		{
			send_boardcast(act, stream_id);
			//开始推流 data是ip列表
			std::string iplist;
			iplist.append((char*)ptrData.get(), length);
			std::cout << "iplist:" << iplist << std::endl;
			auto list1 = nlohmann::json::parse(iplist);
			auto ips = list1["clients"];
			if (mStreamMaps.count(stream_id) < 1)
			{
				mStreamMaps[stream_id] = std::set<std::string>();
			}
			auto& setList = mStreamMaps[stream_id];
			for (auto item : ips)
			{
				setList.emplace(item);
			}
			for (auto item : setList)
			{
				//	std::cout << item <<"will received "<<stream_id << " stream"<< std::endl;
			}
			printStreamMap();
		}
		else if (cmd == 4)
		{
			send_boardcast(act, stream_id);
			//4结束推流
			printStreamMap();
			std::cout << "stop stream " << stream_id << std::endl;
			mStreamMaps.erase(stream_id);
			printStreamMap();
		}
		else {
			send_boardcast(act, stream_id);
		}

		std::cout << "cmd:" << cmd << " stream_id:" << stream_id << " length:" << length << std::endl;

		return true;
	}


private:
	void printStreamMap() {
		std::cout << "streamMapSize:" << mStreamMaps.size() << std::endl;
		for (auto item : mStreamMaps) {
			auto& list = item.second;
			std::string result;
			for (auto ip : list)
			{
				result.append(ip + ";");
			}
			std::cout << "stream:" << item.first << "dispath list " << result << std::endl;
		}
	}

private:
	/**
	 * 发送音频数据
	 */
	void send_boardcast(action& act, const std::string& streamId)
	{

		if (mStreamMaps.count(streamId))
		{
			std::set<std::string>& list = mStreamMaps.at(streamId);
			for (auto& ip : list)
			{
				if (mRegisterList.count(ip))
				{
					connection_hdl& hdl = mRegisterList.at(ip);
					try {
						m_server.send(hdl, act.msg);
					}
					catch (const std::exception& ex)
					{
						std::cout << ex.what() << std::endl;
					}
				}
				else {
					//std::cout << "no support " << std::endl;
				}

			}
		}
		else {
			std::cout << "stream_id:" << streamId << "no groud, not dispatch" << std::endl;
		}
	}

private:
	void time_out_proc() {

		//std::map<connection_hdl, time_t> tempList;
		//std::vector<connection_hdl> needRemoveList;
		//time_t  now = time(0);
		//while (1)
		//{
		//	now = time(0);
		//	needRemoveList.clear();
		//	{
		//	std::lock_guard<std::mutex> lock(mListTimeoutLock);
		//	tempList = mListTimeout;
		//	}

		//	for (auto item : tempList)
		//	{
		//		if (now - item.second > 20)
		//		{
		//			//过期了
		//			needRemoveList.push_back(item.first);
		//		}
		//	}
		//	{
		//		std::lock_guard<std::mutex> lock(mListTimeoutLock);
		//		for (auto item : needRemoveList)
		//		{
		//			if (mListTimeout.count(item) > 0)
		//			{
		//				try{
		//				m_server.close(item, websocketpp::close::status::normal, "timeout");
		//				}
		//				catch (const std::exception& e)
		//				{
		//					std::cout << e.what() << std::endl;
		//				}
		//				mListTimeout.erase(item);
		//			}
		//		}
		//	}

		//	std::this_thread::sleep_for(std::chrono::seconds(5));
		// }
	}

private:

	typedef std::set<connection_hdl, std::owner_less<connection_hdl> > con_list;

	server m_server;
	con_list m_connections;

	//std::map<connection_hdl, time_t> mListTimeout;
	//std::mutex mListTimeoutLock;

	std::queue<action> m_actions;

	mutex m_action_lock;
	mutex m_connection_lock;
	condition_variable m_action_cond;

	/**
	 * 所有注册的ip地址，对应的连接
	 */
	std::map<std::string, connection_hdl> mRegisterList;

	/**
	 * 所有的流，对应的要发送的ip地址
	 */
	std::map<std::string, std::set<std::string>> mStreamMaps;

};

int main()
{
	try {
		std::cout << "compile date:" << __DATE__ << " time:" << __TIME__ << std::endl;
		
		ACCServer server;

		thread t(bind(&ACCServer::process_messages, &server));
		server.run(9999);
		if (t.joinable())
		{
			t.join();
		}

	}
	catch (const std::exception& e)
	{

	}

	std::cout << "pause" << std::endl;
}

// 运行程序: Ctrl + F5 或调试 >“开始执行(不调试)”菜单
// 调试程序: F5 或调试 >“开始调试”菜单

// 入门使用技巧: 
//   1. 使用解决方案资源管理器窗口添加/管理文件
//   2. 使用团队资源管理器窗口连接到源代码管理
//   3. 使用输出窗口查看生成输出和其他消息
//   4. 使用错误列表窗口查看错误
//   5. 转到“项目”>“添加新项”以创建新的代码文件，或转到“项目”>“添加现有项”以将现有代码文件添加到项目
//   6. 将来，若要再次打开此项目，请转到“文件”>“打开”>“项目”并选择 .sln 文件
